
package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.BetweenFormater;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.text.StrSpliter;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.model.MessageCenter;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.async.AsyncTask;
import com.hexinfo.dmpro.sparing.dao.SyncRuleConfMapper;
import com.hexinfo.dmpro.sparing.dto.*;
import com.hexinfo.dmpro.sparing.model.*;
import com.hexinfo.dmpro.sparing.service.*;
import com.hexinfo.dmpro.sparing.util.Logger;
import com.xqfunds.job.core.log.XxlJobLogger;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 金桥到万国同步(库级)
 *
 * @author yemw
 * @date 2023-09-08 17:45:29
 */
@Service
@Slf4j
public class SyncRuleConfServiceDatabaseImpl extends ServiceImpl<SyncRuleConfMapper, SyncRuleConf> implements SyncRuleConfDatabaseService {

    @Value("${hdfs.tarPrefix}")
    private String prefix;
    @Autowired
    private ScanMetadataService scanMetadataService;
    @Autowired
    private HiveToOracleService hiveToOracleService;
    @Autowired
    private MessageCenterService messageCenterService;
    @Autowired
    private AvailabilityPathService availabilityPathService;
    @Autowired
    private DistCPService distCPService;
    @Autowired
    private DataRefreshService dataRefreshService;
    @Autowired
    private SyncCompleteLibraryTaskService syncCompleteLibraryTaskService;
    @Autowired
    private SyncCompleteLibraryService syncCompleteLibraryService;
    @Autowired
    private SyncAuditWhiteListService syncAuditWhiteListService;
    @Autowired
    private AsyncTask asyncTask;
    @Autowired
    private SyncRuleConfService syncRuleConfService;
    @Autowired
    private SyncAuditBlackListService syncAuditBlackListService;
    @Autowired
    private FutureService futureService;

    @Override
    public Map<String,String> getSyncRuleDatabaseData(String database) {
        Map<String,String> map = new HashMap<>();
        //转小写
        database = com.hexinfo.dmpro.common.utils.StringUtils.toLowerCase(database);
        String nameNodeStatus = availabilityPathService.getSouAndTar();
        if (StrUtil.isBlank(nameNodeStatus)){
            log.warn("-------------------高可用节点变动异常");
            map.put("error","高可用节点变动异常");
            return map;
        }
        //消息中心
        MessageCenter mess = new MessageCenter();
        mess.setMenu(ScanCommonConstants.ClusterName.SJRB.value);
        mess.setOperator("admin");
        mess.setDataType(ScanCommonConstants.ClusterName.JXZ.value);
        mess.setMessageSubject(ScanCommonConstants.ClusterName.KJSJTB.value+"~"+database);
        mess.setMessageWeight(ScanCommonConstants.ClusterName.ONE.value);
        messageCenterService.saveOrUpdate(mess);
        HdfsHeatBackup kup =new HdfsHeatBackup();
        DistCPDTO distCPDTO = new DistCPDTO();
        //任务表
        SyncCompleteLibraryTask task = new SyncCompleteLibraryTask();
        //当前开始时间
        Date staDate = DateUtil.date();
        //查询集群信息
        LambdaQueryWrapper<ClusterSparingConf> jQWrapper = Wrappers.lambdaQuery();
        LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
        ClusterSparingConf jQCluster = scanMetadataService.wrapperClusterSparingConf(jQWrapper,ScanCommonConstants.ClusterName.JQ.value);
        ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,ScanCommonConstants.ClusterName.WG.value);
        //稽核所需SQL模板
        String auditSql = "select count(1) as count from {} ";
        //通过sql直接查询元数据库进行全库同步
        List<String> tableNameList = hiveToOracleService.selectHiveComplete(database);
        if(ObjectUtil.isNotEmpty(tableNameList)||tableNameList.size()>0){
            //黑名单
            //不稽核名单过滤 白名单
            List<String> stringList = syncAuditBlackListService.matchList(database,tableNameList,task);
            //忽略名单
            IgnoreListDTO ignoreListDTO = syncAuditBlackListService.ignoreList(database);
            task.setDatabaseName(database);
            task.setTaskName(DateUtil.format(DateUtil.date(), "yyyyMMdd"));
            task.setTaskState(ScanCommonConstants.ClusterName.JXZ.value);
            syncCompleteLibraryTaskService.saveOrUpdate(task);
            //投研系列库才需要自动化建表 aresoft库不执行自动建表
            if(!"aresoft".equals(database)){
                //自动化建表
                automatedTable(database,task);
            }
            //热备记录
            SyncCompleteLibrary libraryRb = new SyncCompleteLibrary();
            libraryRb.setTaskName(task.getTaskName());
            libraryRb.setTaskId(task.getId());
            libraryRb.setLibraryTableName("【"+database+"】同步任务");
            libraryRb.setTaskState(ScanCommonConstants.ClusterName.JXZ.value);
            //HDFS路径拼接
            String  lo = jQCluster.getHdfsPrefixPath() + database + ".db/" ;
            kup.setSourceAddress(lo);
            kup.setTargetAddress(StrUtil.replace(StrUtil.replace(lo, jQCluster.getHdfsPrefixPath(), wGCluster.getHdfsPrefixPath()), "/" + database + ".db/", "/" + prefix + database + ".db/"));
            kup.setSourceDbName(database);
            kup.setTargetDbName(prefix + database);
            distCPDTO.setHdfsHeatBackup(kup);
            libraryRb.setLog("["+DateUtil.now()+"]【同步开始】"+" \n["+DateUtil.now()+"]【源路径】"+kup.getSourceAddress()+" \n["+DateUtil.now()+"]【目标路径】"+ kup.getTargetAddress());
            syncCompleteLibraryService.saveOrUpdate(libraryRb);
            //库级同步
            String err = distCPService.execShellDatabase(distCPDTO,libraryRb);
//            err  = "成功";
//            反洗钱多做一次库级同步
            String aresoftErr = "";
            if("aresoft".equals(database)&&err.contains("成功")){
                HdfsHeatBackup aresoftKup =new HdfsHeatBackup();
                DistCPDTO aresoftDist = new DistCPDTO();
                //HDFS路径拼接
                String  aresoftLo = jQCluster.getHdfsPrefixPath().substring(0,getSlashIndex(jQCluster.getHdfsPrefixPath(),3)) + "/user/aresoft/hadoopData" ;
                aresoftKup.setSourceAddress(aresoftLo);
                aresoftKup.setTargetAddress(wGCluster.getHdfsPrefixPath().substring(0,getSlashIndex(wGCluster.getHdfsPrefixPath(),3)) + "/user/aresoft/hadoopData" );
                aresoftKup.setSourceDbName(database);
                aresoftKup.setTargetDbName(prefix + database);
                aresoftDist.setHdfsHeatBackup(aresoftKup);
                libraryRb.setLog(libraryRb.getLog()+"\n["+DateUtil.now()+"]【aresoft反洗钱同步开始】"+" \n["+DateUtil.now()+"]【源路径】"+aresoftKup.getSourceAddress()+"\n["+DateUtil.now()+"]【目标路径】"+ aresoftKup.getTargetAddress());
                syncCompleteLibraryService.saveOrUpdate(libraryRb);
                aresoftErr = distCPService.execShellDatabase(aresoftDist,libraryRb);
            }
            if(!err.contains("成功")||("aresoft".equals(database)&&!aresoftErr.contains("成功"))){
                task.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                task.setStandbyTime(DateUtil.formatBetween(staDate,DateUtil.date(), BetweenFormater.Level.SECOND));
                syncCompleteLibraryTaskService.saveOrUpdate(task);
                mess.setDataType(ScanCommonConstants.ClusterName.YC.value);
                libraryRb.setLog(libraryRb.getLog()+"\n["+DateUtil.now()+"]【distCP同步异常】"+ err );
                if(StrUtil.isNotBlank(aresoftErr)){
                    libraryRb.setLog(libraryRb.getLog()+"\n["+DateUtil.now()+"]【distCP同步异常】"+ err +"\n["+DateUtil.now()+"]"+ aresoftErr);
                }
                libraryRb.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                libraryRb.setHotStandbyTime(DateUtil.formatBetween(staDate,DateUtil.date(), BetweenFormater.Level.SECOND));
                syncCompleteLibraryService.saveOrUpdate(libraryRb);
                Logger.log(libraryRb.getLog());
                map.put("error","distCP同步异常");
            }else{
                libraryRb.setLog(libraryRb.getLog()+"["+DateUtil.now()+"]【同步成功】");
                libraryRb.setHotStandbyTime(DateUtil.formatBetween(staDate,DateUtil.date(), BetweenFormater.Level.SECOND));
                libraryRb.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
                syncCompleteLibraryService.saveOrUpdate(libraryRb);
                Logger.log(libraryRb.getLog());
                List<Future<Map<String,Object>>> futures = new ArrayList<>();//存线程的返回结果
                for (String tableName: stringList) {
                    //以下调用会启动线程进行执行，多线程并发
                    Future<Map<String,Object>> res = asyncTask.asyncDatabase(dataRefreshService,task,tableName,prefix,auditSql,ignoreListDTO);
                    futures.add(res);
                }
                //会等待所有线程都执行结束，拿到结果
                String errAsync = futureService.futureResult(futures);
                //当前结束时间
                Date endDate = DateUtil.date();
                task.setStandbyTime(DateUtil.formatBetween(staDate,endDate, BetweenFormater.Level.SECOND));
                //无异常
                LambdaQueryWrapper<SyncCompleteLibrary> wrapperSB = Wrappers.lambdaQuery();
                wrapperSB.eq(StrUtil.isNotBlank(task.getId()),SyncCompleteLibrary::getTaskId,task.getId())
                        .eq(SyncCompleteLibrary::getTaskState,ScanCommonConstants.ClusterName.SB.value)
                        .notLike(SyncCompleteLibrary::getLibraryTableName,"自动化建表任务")
                        .eq(SyncCompleteLibrary::getDel,CommonConstants.STATUS_NORMAL);
                List<SyncCompleteLibrary> tableListSB = syncCompleteLibraryService.getBaseMapper().selectList(wrapperSB);
                if(ObjectUtil.isNotEmpty(tableListSB)){
                    task.setTaskState(ScanCommonConstants.ClusterName.SB.value);
                }else{
                    task.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
                }
                //查询日志ID下全部的表信息做源总量和目标总量计算
                //源总数
                Long numSource =  0L;
                //目标总数
                Long numTarget =  0L;
                //日志汇总
                String rz = "["+DateUtil.now()+"]【表操作日志汇总】\n";
                LambdaQueryWrapper<SyncCompleteLibrary> wrapper = Wrappers.lambdaQuery();
                wrapper.eq(StrUtil.isNotBlank(task.getId()),SyncCompleteLibrary::getTaskId,task.getId())
                        .notIn(SyncCompleteLibrary::getTaskState,ScanCommonConstants.ClusterName.HL.value)
                        .eq(SyncCompleteLibrary::getDel,CommonConstants.STATUS_NORMAL);
                List<SyncCompleteLibrary> tableList = syncCompleteLibraryService.getBaseMapper().selectList(wrapper);
                if(ObjectUtil.isNotEmpty(tableList)){
                    for (SyncCompleteLibrary table: tableList) {
                        rz += table.getLog() + "\n";
                        if(ObjectUtil.isNotEmpty(table.getSourceCountNumber())){
                            numSource += table.getSourceCountNumber();
                        }
                        if(ObjectUtil.isNotEmpty(table.getTargetCountNumber())){
                            numTarget += table.getTargetCountNumber();
                        }
                    }
                }
                task.setSourceCountNumber(numSource);
                task.setTargetCountNumber(numTarget);
                syncCompleteLibraryTaskService.saveOrUpdate(task);
                Logger.log(rz);
                if(task.getTaskState().equals(ScanCommonConstants.ClusterName.SB.value)){
                    mess.setDataType(ScanCommonConstants.ClusterName.YC.value);
                    map.put("error","["+DateUtil.now()+"]【同步异常】");
                }else{
                    mess.setDataType(ScanCommonConstants.ClusterName.YWC.value);
                    map.put("success","["+DateUtil.now()+"]【同步完成】");
                }
            }
        }else{
//            Logger.log("【查询表信息】"+database+"库下暂无表信息");
            map.put("error","【查询表信息】"+database+"库下暂无表信息");
            mess.setDataType(ScanCommonConstants.ClusterName.YC.value);
        }
        messageCenterService.saveOrUpdate(mess);
        return map;
    }

    /**
     * 获取第几个斜杆所在位置
     * @return
     */
    public int getSlashIndex(String url,int num){
        int slashCount = 0; // 记录斜杆的数量
        int slashIndex = -1; // 记录第几个斜杆的位置
        for (int i = 0; i < url.length(); i++) {
            if (url.charAt(i) == '/') {
                slashCount++;
                if (slashCount == num) {
                    slashIndex = i;
                    break;
                }
            }
        }
        return slashIndex;
    }

    @Override
    public boolean automatedTable(String dbName,SyncCompleteLibraryTask task){
        Map<String,Integer> intMap = new HashMap<>();
        intMap.put("SB",0);
        intMap.put("YWC",0);
        //需要建表的库表名称
        List<CreateTableDTO> createTableDTOList = syncRuleConfService.createTable(dbName);
        //添加白名单信息
        List<String> whiteList = syncAuditWhiteListService.addWhiteList(dbName);
        if(ObjectUtil.isNotEmpty(whiteList)){
            for (String tableName: whiteList){
                List<String> split = StrUtil.split(tableName,".",-1,true,true);
                if(split.size() == 2){
                    CreateTableDTO createTableDTO = new CreateTableDTO();
                    createTableDTO.setDbTableName(tableName);
                    createTableDTO.setTableName(split.get(1));
                    createTableDTO.setDbName(split.get(0));
                    createTableDTOList.add(createTableDTO);
                }
            }
        }
        SyncCompleteLibrary library = new SyncCompleteLibrary();
        Date staDate = DateUtil.date();
        library.setTaskId(task.getId());
        library.setTaskName(task.getTaskName());
        library.setLibraryTableName("【"+dbName+"】自动化建表任务");
        library.setTaskState(ScanCommonConstants.ClusterName.JXZ.value);
        library.setLog("["+DateUtil.now()+"]【自动化建表开始】");
        library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【所属建表总和】"+createTableDTOList.size());
        library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【表名列表】");
        for (CreateTableDTO tableDTO: createTableDTOList) {
            library.setLog(library.getLog()+"\n"+tableDTO.getDbName()+"."+tableDTO.getTableName());
        }
        syncCompleteLibraryService.saveOrUpdate(library);
        //建表sql集合
        List<Future<Map<String,Object>>> futures = new ArrayList<>();//存线程的返回结果
        for (CreateTableDTO tableDTO: createTableDTOList) {
            //以下调用会启动线程进行执行，多线程并发
            Future<Map<String,Object>> res = asyncTask.forCreateTable(tableDTO,library,intMap);
            futures.add(res);
        }
        //会等待所有线程都执行结束，拿到结果
        futureService.futureResult(futures);
        library.setLog(library.getLog()+"\n["+DateUtil.now()+"]【自动化建表结束】");
        library.setHotStandbyTime(DateUtil.formatBetween(staDate,DateUtil.date(), BetweenFormater.Level.SECOND));
        if(intMap.get("SB")>0){
            library.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            syncCompleteLibraryService.saveOrUpdate(library);
            return false;
        }else{
            library.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
            syncCompleteLibraryService.saveOrUpdate(library);
            return true;
        }
    }
}
